package cn.jly.boot.kafka.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
import org.springframework.messaging.Message;

/**
 * @PackageName cn.jly.boot.kafka.config
 * @ClassName KafkaConfig
 * @Description kafka配置
 * @Author 姬岚洋
 * @Date 2021/1/10 下午 2:42
 */
@Configuration
public class KafkaConfig {

    /**
     * 消息过滤器可以在消息抵达consumer之前被拦截，
     * 在实际应用中，我们可以根据自己的业务逻辑，
     * 筛选出需要的信息再交由KafkaListener处理，
     * 不需要的消息则过滤掉。
     * <p>
     * 配置消息过滤只需要为 监听器工厂 配置一个RecordFilterStrategy（消息过滤策略），
     * 返回true的时候消息将会被抛弃，返回false时，消息能正常抵达监听容器。
     *
     * @return
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory concurrentKafkaListenerContainerFactory(ConsumerFactory consumerFactory) {
        final ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConsumerFactory(consumerFactory);
        // 被过滤的消息被丢弃
        factory.setAckDiscarded(true);
        // 自定义消息过滤策略
        factory.setRecordFilterStrategy(new RecordFilterStrategy() {
            @Override
            public boolean filter(ConsumerRecord consumerRecord) {
                if (consumerRecord.value().toString().contains("中国")) {
                    // 返回true的消息被过滤
                    return true;
                }
                return false;
            }
        });
        return factory;
    }

    /**
     * ConsumerAwareListenerErrorHandler 异常处理器
     * <p>
     * 通过异常处理器，我们可以处理consumer在消费时发生的异常。
     * <p>
     * 新建一个 ConsumerAwareListenerErrorHandler 类型的异常处理方法，用@Bean注入，BeanName默认就是方法名，
     * 然后我们将这个异常处理器的BeanName放到@KafkaListener注解的errorHandler属性里面，当监听抛出异常的时候，则会自动调用异常处理器
     *
     * @return
     */
    @Bean
    public ConsumerAwareListenerErrorHandler consumerAwareListenerErrorHandler() {
        return new ConsumerAwareListenerErrorHandler() {
            @Override
            public Object handleError(Message<?> message, ListenerExecutionFailedException exception, Consumer<?, ?> consumer) {
                System.out.println("消费异常 -> " + message.getPayload());
                return null;
            }
        };
    }

    /**
     * 创建一个topic，分区数目2，副本数目2
     *
     * @return
     */
    @Bean
    public NewTopic initTopic() {
        return new NewTopic("test_from_spring_boot", 2, (short) 2);
    }

    /**
     * 如果要修改分区数，只需修改配置值重启项目即可
     * 修改分区数并不会导致数据的丢失，但是分区数只能增大不能减小
     *
     * @return
     */
    @Bean
    public NewTopic updateTopic() {
        return new NewTopic("test_from_spring_boot", 10, (short) 2);
    }
}
